package com.atguigu.udmp.stream;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.udmp.stream.bean.DimInfo;
import com.atguigu.udmp.stream.bean.DimInfoDefine;
import com.atguigu.udmp.stream.common.function.DorisMapFunction;
import com.atguigu.udmp.stream.common.util.MyDorisSinkUtil;
import com.atguigu.udmp.stream.common.util.MySqlJdbcUtil;

import com.atguigu.udmp.stream.function.DimCheckPartitionFunction;
import com.atguigu.udmp.stream.function.DimInfoExtractProcessFunction;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;
import java.util.stream.Collectors;

public class DimInfoApp {

    public static void main(String[] args) throws Exception {

        // 1  加载要处理的维度表
        // 1.1 先从mysql中加载哪些维度表需要读取
        List<DimInfoDefine> dimInfoList = MySqlJdbcUtil.queryList("select *  from dim_info_define ", DimInfoDefine.class, true);

        List<String> tableList = dimInfoList.stream().map(dimInfo -> dimInfo.getSourceDbName()+"."+dimInfo.getSourceTableName()).collect(Collectors.toList());
        // 1.2根据要加载的表启动flinkcdc
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000L);
        env.setParallelism(4);
        MySqlSource<String> dimSource = MySqlSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .databaseList("gmall_2023_new") // set captured database
                .tableList(tableList.toArray(new String[]{})) // set captured table
                .username("root")
                .password("000000")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .startupOptions(StartupOptions.initial())
                .build();

        DataStreamSource<String> dimJsonStringStrStream = env.fromSource(dimSource, WatermarkStrategy.noWatermarks(), "dim_info").setParallelism(1);
        //
        SingleOutputStreamOperator<JSONObject> dimJsonObjStream = dimJsonStringStrStream.map(jsonString -> JSON.parseObject(jsonString ));

       // dimJsonObjStream.print();
        //2 加载维度的定义信息
        MySqlSource<String> dimInfoDefineSource = MySqlSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .databaseList("udmp") // set captured database
                .tableList("udmp.dim_info_define") // set captured table
                .username("root")
                .password("000000")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .startupOptions(StartupOptions.initial())
                .build();
        DataStreamSource<String> dimInfoDefineJsonStringStream = env.fromSource(dimInfoDefineSource, WatermarkStrategy.noWatermarks(), "dim_info") ;
        //
        SingleOutputStreamOperator<JSONObject> dimInfoDefineJsonObjStream = dimInfoDefineJsonStringStream.map(jsonString -> JSON.parseObject(jsonString ));
        //dimInfoDefineJsonObjStream.print();

        //3 通过维度定义的广播提取维度信息
        MapStateDescriptor mapStateDescriptor = new MapStateDescriptor("dim_info_define",Long.class,DimInfoDefine.class);

        BroadcastStream<JSONObject> dimDefineBroadcastStream = dimInfoDefineJsonObjStream.broadcast(mapStateDescriptor);

        BroadcastConnectedStream<JSONObject, JSONObject> broadcastConnectedStream = dimJsonObjStream.connect(dimDefineBroadcastStream);

        SingleOutputStreamOperator<DimInfo> dimInfoStream = broadcastConnectedStream.process(new DimInfoExtractProcessFunction(mapStateDescriptor));

       // dimInfoStream.print();

        //5 检查并添加分区
        SingleOutputStreamOperator<DimInfo> dimInfoCheckedStream = dimInfoStream.map(new DimCheckPartitionFunction());

        //6 写入doris维度信息表
        SingleOutputStreamOperator<String> dimInfoJsonStringStream = dimInfoCheckedStream.map(new DorisMapFunction<>());


        dimInfoJsonStringStream.sinkTo(MyDorisSinkUtil.getDorisSink("udmp.dim_info","dim_info"));

        env.execute();

    }
}
